/**
 * Free software.
 */

package org.nanhill.commons.pattern.dispatcher;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Producer - Consumer queue use BlockingQueue.
 * 
 * @author alin [xalinx at gmail dot com]
 * @date 2006-11-4
 */
public abstract class BlockingDispatcher<E> implements Dispatcher<E> {
	/**
	 * Blocking queue.
	 */
	private final BlockingQueue<E> queue = new LinkedBlockingQueue<E>();

	/**
	 * Dispatching stop flag.
	 */
	private volatile boolean stopped = false;

	public final void incept(E item) {
		try {
			queue.put(item);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	public void start() {
		new Porter().start();
	}

	public final void stop() {
		stopped = true;
	}

	private class Porter extends Thread {
		public void run() {
			while (true) {
				E workItem = null;
				try {
					workItem = queue.take();
				} catch (InterruptedException e) {
					continue;
				}
				if (stopped)
					break;
				send(workItem);
			}
		}
	}
}
